package com.example.mymessagequeue.mqserver.datacenter;

import com.example.mymessagequeue.common.ConsumerEnv;
import com.example.mymessagequeue.common.MqException;
import com.example.mymessagequeue.mqserver.core.Binding;
import com.example.mymessagequeue.mqserver.core.Exchange;
import com.example.mymessagequeue.mqserver.core.Message;
import com.example.mymessagequeue.mqserver.core.MsgQueue;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用这个类来统一管理内存中的所有数据
 * 该类后续提供的方法, 可能会在多线程环境下被使用, 因此要注意线程安全问题
 */
public class MemoryDataCenter {

    // key: exchangeName, value: exchange  管理交换机
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    // key: queueName, value: msgQueue  管理队列
    private ConcurrentHashMap<String, MsgQueue> queueMap = new ConcurrentHashMap<>();

    // 第一个 key: exchangeName 第二个 key: queueName  管理绑定
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();

    // key: messageId, value: message  管理消息
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();

    // key: queueName, value: List<Message>  队列和消息的关系
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();

    // 第一个 key: queueName 第二个 key: messageId  未确认应答的消息
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // key: consumerTag, value: queueName 表示消费者订阅了哪个队列
    private ConcurrentHashMap<String, String> consumerMap = new ConcurrentHashMap<>();

    /**
     * 消费者下线
     */
    public void consumerOffline(String consumerTag) {
        String queueName = consumerMap.get(consumerTag);
        if (queueName == null) {
            // 该 channel 没有订阅队列
            return;
        }
        MsgQueue queue = getQueue(queueName);
        if (queue == null) {
            // 队列被销毁, 无所谓了
            return;
        }

        // 消费者下线, 队列中管理的消费者 list 中的个数需要移除
        queue.removeConsumerEnv(consumerTag);
        // 消费者与队列绑定的关系需要移除
        consumerMap.remove(consumerTag);
    }

    /**
     * 新增一个消费者订阅某个队列
     */
    public void consumerOnline(ConsumerEnv consumerEnv) {
        // 需要在队列中新增一个消费者订阅
        MsgQueue queue = getQueue(consumerEnv.getQueueName());
        queue.addConsumerEnv(consumerEnv);
        // 需要在内存中管理消费者与队列的关系
        consumerMap.put(consumerEnv.getConsumerTag(), consumerEnv.getQueueName());
    }

    /**
     * 判断消费者是否存活
     * @param consumerTag
     * @return
     */
    public boolean consumerIsAlive(String consumerTag) {
        String queueName = consumerMap.get(consumerTag);
        if (queueName == null) {
            // 该 channel 没有订阅队列
            return false;
        }
        MsgQueue queue = getQueue(queueName);
        if (queue == null) {
            // 队列被销毁
            return false;
        }
        return consumerMap.containsKey(consumerTag);
    }

    // 交换机相关操作

    /**
     * 新增交换机
     *
     * @param exchange
     */
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName=" + exchange.getName());
    }

    /**
     * 查询交换机
     *
     * @param exchangeName
     * @return
     */
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    /**
     * 删除交换机
     *
     * @param exchangeName
     */
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 删除交换机成功! exchangeName=" + exchangeName);
    }

    // 队列相关操作

    /**
     * 新增队列
     *
     * @param queue
     */
    public void insertQueue(MsgQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新增队列成功! queueName=" + queue.getName());
    }

    /**
     * 查询队列
     *
     * @param queueName
     * @return
     */
    public MsgQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    /**
     * 删除队列
     *
     * @param queueName
     */
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 删除队列成功! queueName=" + queueName);
    }


    // 绑定相关操作

    /**
     * 新增一个绑定
     *
     * @param binding
     * @throws MqException
     */
    public void insertBinding(Binding binding) throws MqException {
        // 先使用 exchangeName 查一下, 对应的哈希表是否存在, 不存在则创建
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
//        }

        // 下面的 computeIfAbsent 等价上述的注释代码
        // 如果第一个参数 key 存在, 则找到这个 key 对应的 value, 如果不存在, 则执行第二个参数的 lambda, 并插入到第一个参数的 key 中
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(
                binding.getExchangeName(), k -> new ConcurrentHashMap<>()
        );

        synchronized (bindingMap) {
            // 再根据 queueName 查一下, 如果已经存在绑定了, 就抛出异常, 不存在才能插入
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在 exchangeName=" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新增绑定成功! exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    // 获取绑定, 两个版本
    // 1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    // 2. 根据 exchangeName 获取到所有的 Binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    /**
     * 删除绑定
     *
     * @param binding
     * @throws MqException
     */
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 说明该交换机没有绑定任何队列
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName() +
                    ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 删除绑定成功! exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    // 消息相关操作

    /**
     * 添加消息
     *
     * @param message
     */
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功! messageId=" + message.getMessageId());
    }

    /**
     * 根据 messageId 查找消息
     *
     * @param messageId
     * @return
     */
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    /**
     * 根据 messageId 删除消息
     *
     * @param messageId
     */
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除! messageId=" + messageId);
    }

    /**
     * 发送消息到指定队列
     *
     * @param queue
     * @param message
     */
    public void sendMessage(MsgQueue queue, Message message) {
        // 把消息发送到对应的队列数据结构中
        // 1. 先根据队列名字, 找到对应的消息链表
        // 如果不存在, 则手动创建一个链表 computeIfAbsent 本身就是线程安全的, 是 ConcurrentHashMap 提供的方法

        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        synchronized (message) {
            messages.add(message);
        }
        // 同时也要把消息往总的消息中心插入 (插入到 messageMap 中)
        // 如果 messageMap 已经存在了这个 message 也是没关系的, 主要就是 messageId 相同,
        // 既然 id 相同 message 的内容也一定是一样的, 因为服务器的代码不会对 Message 的内容做修改(basicProperties 和 body)
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中! messageId=" + message.getMessageId());
    }

    /**
     * 从队列中取消息
     * @param queueName
     * @return
     */
    public Message pollMessage(String queueName) {
        // 根据队列名找消息链表, 如果没找到说明队列中没有任何消息
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            // 进行头删, 拿到链表头元素
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出! messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    /**
     * 获取消息元素个数
     * @param queueName
     * @return
     */
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            // 队列中没有消息
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }


    /**
     * 添加未确认的消息
     * @param queueName
     * @param message
     */
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());

        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列! messageId=" + message.getMessageId());
    }

    /**
     * 删除未确认的消息(消息已经确认了)
     * @param queueName
     * @param messageId
     */
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列中删除! messageId=" + messageId);
    }

    /**
     * 获取待确认的消息
     * @param queueName
     * @param messageId
     * @return
     */
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return null;
        }
        return messageHashMap.get(messageId);
    }

    /**
     * 这个方法是从硬盘上读取数据, 把硬盘中之前持久化存储的各个维度的数据都恢复到内存中.
     */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

        // 1. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 2. 恢复所有的队列数据
        List<MsgQueue> queues = diskDataCenter.selectAllQueues();
        for (MsgQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 3. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 4. 恢复所有的消息数据
        // 遍历所有的队列, 拿到每个队列里面的消息
        for (MsgQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());

            // 添加到队列消息映射 map 中
            queueMessageMap.put(queue.getName(), messages);

            // 添加到消息map中
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }

        // 注意！针对 未确认的消息 这部分内存中的数据，不需要从硬盘恢复，之前考虑硬盘存储的时候，也没设定这一块
        // 一旦等待 ack 的过程中，服务器重启了，此时这些 未被确认的消息 就恢复成 未被取走的消息
        // 这个消息在硬盘上存储的时候，就是当作被 未取走的
    }

    /**
     * 查询所有的交换机
     * @return 没有交换机返回空数组
     */
    public List<Exchange> selectAllExchanges() {
        return new ArrayList<>(exchangeMap.values());
    }

    /**
     * 根据交换机名字查找交换机
     */
    public Exchange selectExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    /**
     * 查询所有的队列
     * @return 没有队列返回空的数组
     */
    public List<MsgQueue> selectAllQueues() {
        return new ArrayList<>(queueMap.values());
    }

    /**
     * 查询所有的绑定
     * @return 没有则返回空数组
     */
    public List<Binding> selectAllBindings() {
        List<Map<String, Binding>> list = new ArrayList<>(bindingsMap.values());
        List<Binding> bindings = new ArrayList<>();
        for (Map<String, Binding> item : list) {
            List<Binding> tmpList = new ArrayList<>(item.values());
            bindings.addAll(tmpList);
        }
        return bindings;
    }

    /**
     * 通过队列获取所有的 Message
     * @param queueName
     * @return 没有返回 空数组， 有返回 list
     */
    public List<Message> getMessages(String queueName) {
        List<Message> messages = queueMessageMap.get(queueName);

        if (messages == null) {
            return new ArrayList<>();
        }
        return messages;
    }
}
